package org.wzq;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.Watcher;

/**
 * author:wzq
 * date:16-6-28
 * project:zk_acl
 * description: TODO
 */
public class MqApp2 {

    //    private static final String zkAddress = "104.236.154.96:2181,198.199.108.10:2181,159.203.221.126:2181";
    private static final String zkAddress = "localhost:2181";
    private static final String zkacl_admin = "admin:123456";
    private static final String zkacl_read = "read:123456";

    private static final String sysConfig = "/sysConfig";
    private static final String digest = "digest";
    private static final boolean enableAcl = true;

    private static final boolean isSystem = false;

    private ZkClient zkClient;
    private final String watchPath;
    private final String type;

    public MqApp2(String type) {
        this.type = type;
        this.watchPath = sysConfig + "/" + type;
        this.zkClient = new ZkClient(zkAddress);
        System.out.println("type=" + type);
        System.out.println("watchPath=" + this.watchPath);
    }

    // 权限控制
    // 创建 znode /config if need
    // 应用启动: 注册自己到zk
    // 监听该节点
    public static void main(String[] args) {
        MqApp2 app = new MqApp2("mq");
        app.start();
    }


    public void start() {
        final MqApp2 self = this;
        ZkClient zkClient = new ZkClient(zkAddress);

        if (enableAcl) zkClient.addAuthInfo(digest, zkacl_read.getBytes()); // acl

        // create config if need
        if (!zkClient.exists(sysConfig)) {
            zkClient.createPersistent(sysConfig);
        }
        if (!zkClient.exists(watchPath)) {
            zkClient.createPersistent(watchPath);
        }

        // listen watchPath
        zkClient.subscribeDataChanges(this.watchPath, new IZkDataListener() {
            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println("handleDataChange, path=" + s + "Object=" + o.toString());
                if (self.listener != null) self.listener.onConfigChange((String) o);
            }

            public void handleDataDeleted(String s) throws Exception {
                System.out.println("handleDataDeleted");
                System.out.println("path=" + s);
                if (self.listener != null) self.listener.onConfigDelete();
            }
        });

        // 开始监听 sysConfig 节点
        zkClient.subscribeStateChanges(new IZkStateListener() {
            public void handleStateChanged(Watcher.Event.KeeperState keeperState) throws Exception {
                System.out.println("subscribeStateChanges handleStateChanged");
                System.out.println("keeperState=" + keeperState);
            }

            public void handleNewSession() throws Exception {
                System.out.println("subscribeStateChanges handleStateChanged");
            }

            public void handleSessionEstablishmentError(Throwable throwable) throws Exception {
                System.out.println("subscribeStateChanges handleSessionEstablishmentError");
            }
        });

        registerListener(new ConfigChangeListener() {
            public void onConfigChange(String newCfg) {
                System.out.println("ConfigChangeListener::onConfigChange");
            }

            public void onConfigDelete() {
                System.out.println("ConfigChangeListener::onConfigDelete");
            }
        });

        long sleepTime = 500;
        while (true) {
            try {
                Thread.sleep(sleepTime);
                // String currData = zkClient.readData(watchPath);
                // System.out.println("..." + currData);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
                sleepTime += 1000;
            }
        }
    }

    public interface ConfigChangeListener {
        void onConfigChange(String newCfg);

        void onConfigDelete();
    }

    ConfigChangeListener listener = null;

    public void registerListener(ConfigChangeListener listener) {
        this.listener = listener;
    }


}
